@Weijun-H has been working on this with the Synnada team for a while. The initial benchmark results were promising, so we decided to continue development while receiving community feedback 🚀

This is still in somewhat early stages, and there is work to do. But it might be good to get feedback early on from the community as the performance of this code is somewhat sensitive to idioms used with channels etc.

This is still in somewhat early stages, and there is work to do. But it might be good to get feedback early on from the community as the performance of this code is somewhat sensitive to idioms used with channels etc.

Maybe I am missing something, but the benchmark numbers reported above don't really show much of an improvement

For example, this branch appears to be basically the same

│ Total Time (main)                                │ 11767.43ms │
│ Total Time (on-demand-repartition-with-config)   │ 11787.12ms │

Are there any benchmarks that show a performance benefit of all this new code?

@Weijun-H did some benchmarks a while back and the approach seemed promising in TPCH/SF50.

@mertak-synnada will do a detailed review of this tomorrow and then @Weijun-H can run the latest benchmarks for us to see how the performance changes

Maybe I am missing something, but the benchmark numbers reported above don't really show much of an improvement

this might be a silly question but, did you set the config flag for on-demand-repartition-with-config branch?😅

I updated the latest benchmark results. It appears that the OnDemandRepartition improved performance on clickbench_partitioned and large datasets such as tpch_sf50. For tpch_sf1 and tpch_sf10, the results are similar. I will check again to ensure everything is functioning correctly in the coming days.
@ozankabak @alamb @berkaysynnada @mertak-synnada

UPDATE:
I reviewed this PR again and verified that the optimizer's behavior remains consistent with RoundRobinBatch after incorporating OnDemandRepartitionExec. Based on the current results, performance is comparable for tpch and tpch_10. Additionally, some queries in clickbench_partitioned benefit from this change. This PR also reduces excessive memory usage caused by prefetching.

Impressive work! I got a suggestion and a high-level question:

Suggestion

I think to justify this change, we have to make sure:

• No performance regression (benchmarks already showed)
• Reduce memory footprint, for queries which batch can accumulate in RepartitionExec (as the origin issue said)

I tried to check the memory usage for tpch-sf10 and clickbench, there is no noticeable change for those queries.
Perhaps we should construct queries with this anti-pattern, and demonstrate memory usage can actually reduced by this on-demand repartition executor?

Here is a script for checking memory usage in benchmark queries

# This script should be placed under benchmarks/
#
# Supported benchmarks are 'tpch' and 'clickbench'
#
# Example usage:
# Run TPCH benchmark and save results:
#   python3 membench.py run --benchmark tpch --result tpch_main.csv
#   python3 membench.py run --benchmark tpch --result tpch_optimized.csv
#
# Compare results:
#   python3 membench.py compare tpch_main.csv tpch_optimized.csv

import subprocess
import re
import csv
import argparse

def human_readable_size(size):
    units = ["B", "K", "M", "G", "T"]
    index = 0
    while size >= 1024 and index < len(units) - 1:
        size /= 1024.0
        index += 1
    return f"{size:.2f}{units[index]}"

def run_tpch_queries(label, result_file):
    results = []
    for query in range(1, 23):
        cmd = [
            "/usr/bin/time", "-l", "cargo", "run", "--release", "--bin", "dfbench", 
            "tpch", "--format", "parquet", "--path", "./data/tpch_sf10", 
            "--query", str(query), "--iterations", "1"
        ]
        
        process = subprocess.run(cmd, capture_output=True, text=True, shell=False)
        stderr_output = process.stderr
        
        match = re.search(r"(\d+)\s+maximum resident set size", stderr_output)
        max_rss = human_readable_size(int(match.group(1))) if match else "N/A"
        results.append((query, max_rss))
    
    with open(result_file, "w", newline='') as f:
        writer = csv.writer(f)
        writer.writerow(["Query", "Memory"])
        writer.writerows(results)
    
    print(f"Results saved to {result_file}")

def run_clickbench_queries(label, result_file):
    results = []
    for query in range(0, 43):
        cmd = [
            "/usr/bin/time", "-l", "cargo", "run", "--release", "--bin", "dfbench", 
            "clickbench", "--path", "./data/hits.parquet", 
            "--queries-path", "./queries/clickbench/queries.sql", 
            "--query", str(query), "--iterations", "1"
        ]
        
        process = subprocess.run(cmd, capture_output=True, text=True, shell=False)
        stderr_output = process.stderr
        
        match = re.search(r"(\d+)\s+maximum resident set size", stderr_output)
        max_rss = human_readable_size(int(match.group(1))) if match else "N/A"
        results.append((query, max_rss))
    
    with open(result_file, "w", newline='') as f:
        writer = csv.writer(f)
        writer.writerow(["Query", "Memory"])
        writer.writerows(results)
    
    print(f"Results saved to {result_file}")

def compare_results(file1, file2):
    results1, results2 = {}, {}
    
    with open(file1, "r") as f1, open(file2, "r") as f2:
        reader1, reader2 = csv.reader(f1), csv.reader(f2)
        next(reader1)  # Skip header
        next(reader2)  # Skip header
        
        for row in reader1:
            results1[row[0]] = row[1]
        for row in reader2:
            results2[row[0]] = row[1]
    
    print(f"{'Query':<10}{'Branch1':<10}{'Branch2':<10}{'Change'}")
    for query in results1:
        mem1 = results1[query]
        mem2 = results2.get(query, "N/A")
        
        if mem1 != "N/A" and mem2 != "N/A":
            size1 = float(mem1[:-1])
            size2 = float(mem2[:-1])
            ratio = size2 / size1 if size1 > 0 else 1.0
            change = f"{ratio:.2f}X" if abs(ratio - 1) > 0.05 else "No Change"
        else:
            change = "N/A"
        
        print(f"{query:<10}{mem1:<10}{mem2:<10}{change}")

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("mode", choices=["run", "compare"], help="Run benchmarks or compare results")
    parser.add_argument("--result", help="Output result file for benchmarking")
    parser.add_argument("--benchmark", choices=["tpch", "clickbench"], help="Specify which benchmark to run")
    parser.add_argument("file1", nargs="?", help="First result file for comparison")
    parser.add_argument("file2", nargs="?", help="Second result file for comparison")
    args = parser.parse_args()
    
    if args.mode == "run" and args.result and args.benchmark:
        if args.benchmark == "tpch":
            run_tpch_queries("run", args.result)
        elif args.benchmark == "clickbench":
            run_clickbench_queries("run", args.result)
    elif args.mode == "compare" and args.file1 and args.file2:
        compare_results(args.file1, args.file2)
    else:
        print("Invalid arguments. Use --help for usage information.")

if __name__ == "__main__":
    main()

Results:

TPCH:
----
Query     Branch1   Branch2   Change
1         464.05M   460.78M   No Change
2         397.00M   412.77M   No Change
3         714.56M   630.64M   0.88X
4         408.53M   418.78M   No Change
5         741.30M   769.73M   No Change
6         390.02M   398.72M   No Change
7         3.41G     3.45G     No Change
8         1.08G     1.05G     No Change
9         2.37G     2.31G     No Change
10        1.11G     1.16G     No Change
11        260.78M   267.41M   No Change
12        429.95M   449.06M   No Change
13        675.67M   668.22M   No Change
14        666.56M   700.22M   No Change
15        673.66M   656.70M   No Change
16        485.81M   474.59M   No Change
17        605.38M   631.92M   No Change
18        3.26G     3.29G     No Change
19        500.77M   577.95M   1.15X
20        1.07G     1.05G     No Change
21        982.59M   978.69M   No Change
22        303.86M   302.14M   No Change

Clickbench:
...(no change)

Question

In my understanding the new repartition executor is a wrapper on RepartitionExec, to enable lazy evaluation, it should support both RoundRobin and Hash repartition right? This PR only swapped RoundRobin, do you also plan to add on-demand hash repartition in the future?

I ran some tests yesterday and I can confirm the runtime improvements.
I do get some high memory usage however especially with some queries (TPC-H Query 18 I believe) than when using round-robin repartitioning. Are there some ways to get it down (e.g. use bounded channels or otherwise?)

I ran some tests yesterday and I can confirm the runtime improvements. I do get some high memory usage however especially with some queries (TPC-H Query 18 I believe) than when using round-robin repartitioning. Are there some ways to get it down (e.g. use bounded channels or otherwise?)

I tried to avoid using yield_now when waiting for the child operator data; this should lower memory usage.
benchmark after adopting this approach

The performance decreased in many cases @Dandandan

--------------------
Benchmark clickbench_partitioned.json
--------------------
┏━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query        ┃ on-demand-repartition-with-config ┃ on-demand-not-always-add-roundrobin ┃        Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 0     │                            1.36ms │                              1.44ms │  1.05x slower │
│ QQuery 1     │                           30.28ms │                             21.71ms │ +1.39x faster │
│ QQuery 2     │                           73.36ms │                             64.39ms │ +1.14x faster │
│ QQuery 3     │                           64.40ms │                             52.36ms │ +1.23x faster │
│ QQuery 4     │                          512.46ms │                            486.49ms │ +1.05x faster │
│ QQuery 5     │                          571.75ms │                            539.88ms │ +1.06x faster │
│ QQuery 6     │                           31.22ms │                             21.57ms │ +1.45x faster │
│ QQuery 7     │                           33.53ms │                             25.52ms │ +1.31x faster │
│ QQuery 8     │                          568.44ms │                            532.79ms │ +1.07x faster │
│ QQuery 9     │                          758.40ms │                            766.04ms │     no change │
│ QQuery 10    │                          180.18ms │                            170.90ms │ +1.05x faster │
│ QQuery 11    │                          202.26ms │                            184.72ms │ +1.09x faster │
│ QQuery 12    │                          601.67ms │                            578.79ms │     no change │
│ QQuery 13    │                          832.37ms │                            802.30ms │     no change │
│ QQuery 14    │                          581.32ms │                            528.55ms │ +1.10x faster │
│ QQuery 15    │                          625.30ms │                            590.98ms │ +1.06x faster │
│ QQuery 16    │                         1362.52ms │                           1162.10ms │ +1.17x faster │
│ QQuery 17    │                         1258.20ms │                           1107.68ms │ +1.14x faster │
│ QQuery 18    │                         3628.57ms │                           3589.80ms │     no change │
│ QQuery 19    │                           59.32ms │                             48.46ms │ +1.22x faster │
│ QQuery 20    │                          884.00ms │                            817.40ms │ +1.08x faster │
│ QQuery 21    │                         1066.61ms │                           1044.21ms │     no change │
│ QQuery 22    │                         1883.62ms │                           2261.84ms │  1.20x slower │
│ QQuery 23    │                         6562.12ms │                           6412.42ms │     no change │
│ QQuery 24    │                          337.47ms │                            329.45ms │     no change │
│ QQuery 25    │                          263.68ms │                            284.84ms │  1.08x slower │
│ QQuery 26    │                          366.64ms │                            359.23ms │     no change │
│ QQuery 27    │                         1185.09ms │                           1180.38ms │     no change │
│ QQuery 28    │                         8860.10ms │                           9427.73ms │  1.06x slower │
│ QQuery 29    │                          441.28ms │                            466.33ms │  1.06x slower │
│ QQuery 30    │                          625.09ms │                            585.79ms │ +1.07x faster │
│ QQuery 31    │                          552.68ms │                            587.15ms │  1.06x slower │
│ QQuery 32    │                         4358.41ms │                           4224.30ms │     no change │
│ QQuery 33    │                         6845.44ms │                           4678.45ms │ +1.46x faster │
│ QQuery 34    │                         7866.05ms │                           4627.91ms │ +1.70x faster │
│ QQuery 35    │                          841.92ms │                            802.52ms │     no change │
│ QQuery 36    │                           80.70ms │                            103.95ms │  1.29x slower │
│ QQuery 37    │                           33.40ms │                             48.88ms │  1.46x slower │
│ QQuery 38    │                           68.14ms │                             70.24ms │     no change │
│ QQuery 39    │                          144.39ms │                            188.87ms │  1.31x slower │
│ QQuery 40    │                           20.52ms │                             22.32ms │  1.09x slower │
│ QQuery 41    │                           20.79ms │                             20.35ms │     no change │
│ QQuery 42    │                           17.83ms │                             27.16ms │  1.52x slower │
└──────────────┴───────────────────────────────────┴─────────────────────────────────────┴───────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┓
┃ Benchmark Summary                                  ┃            ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━┩
│ Total Time (on-demand-repartition-with-config)     │ 55302.89ms │
│ Total Time (on-demand-not-always-add-roundrobin)   │ 49848.22ms │
│ Average Time (on-demand-repartition-with-config)   │  1286.11ms │
│ Average Time (on-demand-not-always-add-roundrobin) │  1159.26ms │
│ Queries Faster                                     │         19 │
│ Queries Slower                                     │         11 │
│ Queries with No Change                             │         13 │
└────────────────────────────────────────────────────┴────────────┘
--------------------
Benchmark tpch_mem_sf1.json
--------------------
┏━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query        ┃ on-demand-repartition-with-config ┃ on-demand-not-always-add-roundrobin ┃        Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 1     │                           85.39ms │                             71.16ms │ +1.20x faster │
│ QQuery 2     │                           14.22ms │                             13.07ms │ +1.09x faster │
│ QQuery 3     │                           24.57ms │                             22.43ms │ +1.10x faster │
│ QQuery 4     │                           15.57ms │                             11.08ms │ +1.40x faster │
│ QQuery 5     │                           41.50ms │                             34.96ms │ +1.19x faster │
│ QQuery 6     │                            4.82ms │                              4.25ms │ +1.14x faster │
│ QQuery 7     │                           74.88ms │                             66.79ms │ +1.12x faster │
│ QQuery 8     │                           17.29ms │                             15.53ms │ +1.11x faster │
│ QQuery 9     │                           41.62ms │                             37.89ms │ +1.10x faster │
│ QQuery 10    │                           35.49ms │                             32.30ms │ +1.10x faster │
│ QQuery 11    │                            6.85ms │                              5.75ms │ +1.19x faster │
│ QQuery 12    │                           24.91ms │                             20.74ms │ +1.20x faster │
│ QQuery 13    │                           17.21ms │                             16.67ms │     no change │
│ QQuery 14    │                            5.16ms │                              5.06ms │     no change │
│ QQuery 15    │                           12.00ms │                             11.47ms │     no change │
│ QQuery 16    │                           12.75ms │                             12.75ms │     no change │
│ QQuery 17    │                           57.71ms │                             56.38ms │     no change │
│ QQuery 18    │                          124.63ms │                            121.32ms │     no change │
│ QQuery 19    │                           24.50ms │                             24.22ms │     no change │
│ QQuery 20    │                           20.86ms │                             20.35ms │     no change │
│ QQuery 21    │                           86.02ms │                             84.99ms │     no change │
│ QQuery 22    │                           18.71ms │                             18.49ms │     no change │
└──────────────┴───────────────────────────────────┴─────────────────────────────────────┴───────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━┓
┃ Benchmark Summary                                  ┃          ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━┩
│ Total Time (on-demand-repartition-with-config)     │ 766.66ms │
│ Total Time (on-demand-not-always-add-roundrobin)   │ 707.64ms │
│ Average Time (on-demand-repartition-with-config)   │  34.85ms │
│ Average Time (on-demand-not-always-add-roundrobin) │  32.17ms │
│ Queries Faster                                     │       12 │
│ Queries Slower                                     │        0 │
│ Queries with No Change                             │       10 │
└────────────────────────────────────────────────────┴──────────┘
--------------------
Benchmark tpch_mem_sf10.json
--------------------
┏━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━┓
┃ Query        ┃ on-demand-repartition-with-config ┃ on-demand-not-always-add-roundrobin ┃         Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━┩
│ QQuery 1     │                         1335.99ms │                           1342.40ms │      no change │
│ QQuery 2     │                          121.22ms │                            131.85ms │   1.09x slower │
│ QQuery 3     │                          250.98ms │                            286.30ms │   1.14x slower │
│ QQuery 4     │                          125.97ms │                            121.37ms │      no change │
│ QQuery 5     │                          766.43ms │                            661.07ms │  +1.16x faster │
│ QQuery 6     │                          474.20ms │                            317.05ms │  +1.50x faster │
│ QQuery 7     │                         1190.87ms │                           1556.33ms │   1.31x slower │
│ QQuery 8     │                          584.23ms │                            743.68ms │   1.27x slower │
│ QQuery 9     │                         1150.48ms │                           1581.38ms │   1.37x slower │
│ QQuery 10    │                          898.01ms │                            875.53ms │      no change │
│ QQuery 11    │                          113.16ms │                            115.31ms │      no change │
│ QQuery 12    │                          321.89ms │                            665.48ms │   2.07x slower │
│ QQuery 13    │                          340.09ms │                            350.92ms │      no change │
│ QQuery 14    │                          566.53ms │                             49.00ms │ +11.56x faster │
│ QQuery 15    │                          123.92ms │                            159.39ms │   1.29x slower │
│ QQuery 16    │                           93.89ms │                            104.72ms │   1.12x slower │
│ QQuery 17    │                          912.10ms │                            842.42ms │  +1.08x faster │
│ QQuery 18    │                         4680.46ms │                           4226.43ms │  +1.11x faster │
│ QQuery 19    │                          857.78ms │                            825.55ms │      no change │
│ QQuery 20    │                          241.99ms │                            353.86ms │   1.46x slower │
│ QQuery 21    │                         1915.38ms │                           1949.02ms │      no change │
│ QQuery 22    │                           90.42ms │                             98.60ms │   1.09x slower │
└──────────────┴───────────────────────────────────┴─────────────────────────────────────┴────────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┓
┃ Benchmark Summary                                  ┃            ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━┩
│ Total Time (on-demand-repartition-with-config)     │ 17155.98ms │
│ Total Time (on-demand-not-always-add-roundrobin)   │ 17357.64ms │
│ Average Time (on-demand-repartition-with-config)   │   779.82ms │
│ Average Time (on-demand-not-always-add-roundrobin) │   788.98ms │
│ Queries Faster                                     │          5 │
│ Queries Slower                                     │         10 │
│ Queries with No Change                             │          7 │
└────────────────────────────────────────────────────┴────────────┘
--------------------
Benchmark tpch_sf1.json
--------------------
┏━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query        ┃ on-demand-repartition-with-config ┃ on-demand-not-always-add-roundrobin ┃        Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 1     │                          106.51ms │                             95.95ms │ +1.11x faster │
│ QQuery 2     │                           20.61ms │                             19.91ms │     no change │
│ QQuery 3     │                           43.59ms │                             36.34ms │ +1.20x faster │
│ QQuery 4     │                           26.63ms │                             22.75ms │ +1.17x faster │
│ QQuery 5     │                           66.74ms │                             54.61ms │ +1.22x faster │
│ QQuery 6     │                           20.47ms │                             17.84ms │ +1.15x faster │
│ QQuery 7     │                           84.02ms │                             74.00ms │ +1.14x faster │
│ QQuery 8     │                           54.67ms │                             49.70ms │ +1.10x faster │
│ QQuery 9     │                           78.47ms │                             66.34ms │ +1.18x faster │
│ QQuery 10    │                           67.30ms │                             59.63ms │ +1.13x faster │
│ QQuery 11    │                           16.11ms │                             14.86ms │ +1.08x faster │
│ QQuery 12    │                           42.73ms │                             33.38ms │ +1.28x faster │
│ QQuery 13    │                           38.63ms │                             32.24ms │ +1.20x faster │
│ QQuery 14    │                           31.97ms │                             29.82ms │ +1.07x faster │
│ QQuery 15    │                           47.89ms │                             42.21ms │ +1.13x faster │
│ QQuery 16    │                           16.23ms │                             14.64ms │ +1.11x faster │
│ QQuery 17    │                          103.78ms │                             95.82ms │ +1.08x faster │
│ QQuery 18    │                          134.12ms │                            117.78ms │ +1.14x faster │
│ QQuery 19    │                           52.97ms │                             48.17ms │ +1.10x faster │
│ QQuery 20    │                           46.33ms │                             39.97ms │ +1.16x faster │
│ QQuery 21    │                          110.85ms │                             93.35ms │ +1.19x faster │
│ QQuery 22    │                           18.11ms │                             16.01ms │ +1.13x faster │
└──────────────┴───────────────────────────────────┴─────────────────────────────────────┴───────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━┓
┃ Benchmark Summary                                  ┃           ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━┩
│ Total Time (on-demand-repartition-with-config)     │ 1228.72ms │
│ Total Time (on-demand-not-always-add-roundrobin)   │ 1075.33ms │
│ Average Time (on-demand-repartition-with-config)   │   55.85ms │
│ Average Time (on-demand-not-always-add-roundrobin) │   48.88ms │
│ Queries Faster                                     │        21 │
│ Queries Slower                                     │         0 │
│ Queries with No Change                             │         1 │
└────────────────────────────────────────────────────┴───────────┘
--------------------
Benchmark tpch_sf10.json
--------------------
┏━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query        ┃ on-demand-repartition-with-config ┃ on-demand-not-always-add-roundrobin ┃        Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 1     │                          843.22ms │                            943.27ms │  1.12x slower │
│ QQuery 2     │                          123.24ms │                            136.28ms │  1.11x slower │
│ QQuery 3     │                          407.30ms │                            446.22ms │  1.10x slower │
│ QQuery 4     │                          198.17ms │                            231.51ms │  1.17x slower │
│ QQuery 5     │                          604.48ms │                            671.58ms │  1.11x slower │
│ QQuery 6     │                          136.38ms │                            158.40ms │  1.16x slower │
│ QQuery 7     │                          887.74ms │                            967.93ms │  1.09x slower │
│ QQuery 8     │                          628.50ms │                            698.72ms │  1.11x slower │
│ QQuery 9     │                         1009.86ms │                           1097.19ms │  1.09x slower │
│ QQuery 10    │                          570.20ms │                            559.98ms │     no change │
│ QQuery 11    │                           90.48ms │                             88.46ms │     no change │
│ QQuery 12    │                          299.50ms │                            282.05ms │ +1.06x faster │
│ QQuery 13    │                          421.75ms │                            421.93ms │     no change │
│ QQuery 14    │                          231.66ms │                            231.48ms │     no change │
│ QQuery 15    │                          384.69ms │                            412.02ms │  1.07x slower │
│ QQuery 16    │                           96.85ms │                             96.86ms │     no change │
│ QQuery 17    │                         1088.84ms │                           1088.24ms │     no change │
│ QQuery 18    │                         1874.24ms │                           1587.75ms │ +1.18x faster │
│ QQuery 19    │                          462.80ms │                            395.33ms │ +1.17x faster │
│ QQuery 20    │                          429.06ms │                            378.36ms │ +1.13x faster │
│ QQuery 21    │                         1564.34ms │                           1344.72ms │ +1.16x faster │
│ QQuery 22    │                          144.52ms │                            130.80ms │ +1.10x faster │
└──────────────┴───────────────────────────────────┴─────────────────────────────────────┴───────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┓
┃ Benchmark Summary                                  ┃            ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━┩
│ Total Time (on-demand-repartition-with-config)     │ 12497.84ms │
│ Total Time (on-demand-not-always-add-roundrobin)   │ 12369.07ms │
│ Average Time (on-demand-repartition-with-config)   │   568.08ms │
│ Average Time (on-demand-not-always-add-roundrobin) │   562.23ms │
│ Queries Faster                                     │          6 │
│ Queries Slower                                     │         10 │
│ Queries with No Change                             │          6 │
└────────────────────────────────────────────────────┴────────────┘

I ran some tests yesterday and I can confirm the runtime improvements. I do get some high memory usage however especially with some queries (TPC-H Query 18 I believe) than when using round-robin repartitioning. Are there some ways to get it down (e.g. use bounded channels or otherwise?)

I tried to avoid using yield_now when waiting for the child operator data; this should lower memory usage. benchmark after adopting this approach

The performance decreased in many cases @Dandandan

--------------------
Benchmark clickbench_partitioned.json
--------------------
┏━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query        ┃ on-demand-repartition-with-config ┃ on-demand-not-always-add-roundrobin ┃        Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 0     │                            1.36ms │                              1.44ms │  1.05x slower │
│ QQuery 1     │                           30.28ms │                             21.71ms │ +1.39x faster │
│ QQuery 2     │                           73.36ms │                             64.39ms │ +1.14x faster │
│ QQuery 3     │                           64.40ms │                             52.36ms │ +1.23x faster │
│ QQuery 4     │                          512.46ms │                            486.49ms │ +1.05x faster │
│ QQuery 5     │                          571.75ms │                            539.88ms │ +1.06x faster │
│ QQuery 6     │                           31.22ms │                             21.57ms │ +1.45x faster │
│ QQuery 7     │                           33.53ms │                             25.52ms │ +1.31x faster │
│ QQuery 8     │                          568.44ms │                            532.79ms │ +1.07x faster │
│ QQuery 9     │                          758.40ms │                            766.04ms │     no change │
│ QQuery 10    │                          180.18ms │                            170.90ms │ +1.05x faster │
│ QQuery 11    │                          202.26ms │                            184.72ms │ +1.09x faster │
│ QQuery 12    │                          601.67ms │                            578.79ms │     no change │
│ QQuery 13    │                          832.37ms │                            802.30ms │     no change │
│ QQuery 14    │                          581.32ms │                            528.55ms │ +1.10x faster │
│ QQuery 15    │                          625.30ms │                            590.98ms │ +1.06x faster │
│ QQuery 16    │                         1362.52ms │                           1162.10ms │ +1.17x faster │
│ QQuery 17    │                         1258.20ms │                           1107.68ms │ +1.14x faster │
│ QQuery 18    │                         3628.57ms │                           3589.80ms │     no change │
│ QQuery 19    │                           59.32ms │                             48.46ms │ +1.22x faster │
│ QQuery 20    │                          884.00ms │                            817.40ms │ +1.08x faster │
│ QQuery 21    │                         1066.61ms │                           1044.21ms │     no change │
│ QQuery 22    │                         1883.62ms │                           2261.84ms │  1.20x slower │
│ QQuery 23    │                         6562.12ms │                           6412.42ms │     no change │
│ QQuery 24    │                          337.47ms │                            329.45ms │     no change │
│ QQuery 25    │                          263.68ms │                            284.84ms │  1.08x slower │
│ QQuery 26    │                          366.64ms │                            359.23ms │     no change │
│ QQuery 27    │                         1185.09ms │                           1180.38ms │     no change │
│ QQuery 28    │                         8860.10ms │                           9427.73ms │  1.06x slower │
│ QQuery 29    │                          441.28ms │                            466.33ms │  1.06x slower │
│ QQuery 30    │                          625.09ms │                            585.79ms │ +1.07x faster │
│ QQuery 31    │                          552.68ms │                            587.15ms │  1.06x slower │
│ QQuery 32    │                         4358.41ms │                           4224.30ms │     no change │
│ QQuery 33    │                         6845.44ms │                           4678.45ms │ +1.46x faster │
│ QQuery 34    │                         7866.05ms │                           4627.91ms │ +1.70x faster │
│ QQuery 35    │                          841.92ms │                            802.52ms │     no change │
│ QQuery 36    │                           80.70ms │                            103.95ms │  1.29x slower │
│ QQuery 37    │                           33.40ms │                             48.88ms │  1.46x slower │
│ QQuery 38    │                           68.14ms │                             70.24ms │     no change │
│ QQuery 39    │                          144.39ms │                            188.87ms │  1.31x slower │
│ QQuery 40    │                           20.52ms │                             22.32ms │  1.09x slower │
│ QQuery 41    │                           20.79ms │                             20.35ms │     no change │
│ QQuery 42    │                           17.83ms │                             27.16ms │  1.52x slower │
└──────────────┴───────────────────────────────────┴─────────────────────────────────────┴───────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┓
┃ Benchmark Summary                                  ┃            ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━┩
│ Total Time (on-demand-repartition-with-config)     │ 55302.89ms │
│ Total Time (on-demand-not-always-add-roundrobin)   │ 49848.22ms │
│ Average Time (on-demand-repartition-with-config)   │  1286.11ms │
│ Average Time (on-demand-not-always-add-roundrobin) │  1159.26ms │
│ Queries Faster                                     │         19 │
│ Queries Slower                                     │         11 │
│ Queries with No Change                             │         13 │
└────────────────────────────────────────────────────┴────────────┘
--------------------
Benchmark tpch_mem_sf1.json
--------------------
┏━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query        ┃ on-demand-repartition-with-config ┃ on-demand-not-always-add-roundrobin ┃        Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 1     │                           85.39ms │                             71.16ms │ +1.20x faster │
│ QQuery 2     │                           14.22ms │                             13.07ms │ +1.09x faster │
│ QQuery 3     │                           24.57ms │                             22.43ms │ +1.10x faster │
│ QQuery 4     │                           15.57ms │                             11.08ms │ +1.40x faster │
│ QQuery 5     │                           41.50ms │                             34.96ms │ +1.19x faster │
│ QQuery 6     │                            4.82ms │                              4.25ms │ +1.14x faster │
│ QQuery 7     │                           74.88ms │                             66.79ms │ +1.12x faster │
│ QQuery 8     │                           17.29ms │                             15.53ms │ +1.11x faster │
│ QQuery 9     │                           41.62ms │                             37.89ms │ +1.10x faster │
│ QQuery 10    │                           35.49ms │                             32.30ms │ +1.10x faster │
│ QQuery 11    │                            6.85ms │                              5.75ms │ +1.19x faster │
│ QQuery 12    │                           24.91ms │                             20.74ms │ +1.20x faster │
│ QQuery 13    │                           17.21ms │                             16.67ms │     no change │
│ QQuery 14    │                            5.16ms │                              5.06ms │     no change │
│ QQuery 15    │                           12.00ms │                             11.47ms │     no change │
│ QQuery 16    │                           12.75ms │                             12.75ms │     no change │
│ QQuery 17    │                           57.71ms │                             56.38ms │     no change │
│ QQuery 18    │                          124.63ms │                            121.32ms │     no change │
│ QQuery 19    │                           24.50ms │                             24.22ms │     no change │
│ QQuery 20    │                           20.86ms │                             20.35ms │     no change │
│ QQuery 21    │                           86.02ms │                             84.99ms │     no change │
│ QQuery 22    │                           18.71ms │                             18.49ms │     no change │
└──────────────┴───────────────────────────────────┴─────────────────────────────────────┴───────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━┓
┃ Benchmark Summary                                  ┃          ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━┩
│ Total Time (on-demand-repartition-with-config)     │ 766.66ms │
│ Total Time (on-demand-not-always-add-roundrobin)   │ 707.64ms │
│ Average Time (on-demand-repartition-with-config)   │  34.85ms │
│ Average Time (on-demand-not-always-add-roundrobin) │  32.17ms │
│ Queries Faster                                     │       12 │
│ Queries Slower                                     │        0 │
│ Queries with No Change                             │       10 │
└────────────────────────────────────────────────────┴──────────┘
--------------------
Benchmark tpch_mem_sf10.json
--------------------
┏━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━┓
┃ Query        ┃ on-demand-repartition-with-config ┃ on-demand-not-always-add-roundrobin ┃         Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━┩
│ QQuery 1     │                         1335.99ms │                           1342.40ms │      no change │
│ QQuery 2     │                          121.22ms │                            131.85ms │   1.09x slower │
│ QQuery 3     │                          250.98ms │                            286.30ms │   1.14x slower │
│ QQuery 4     │                          125.97ms │                            121.37ms │      no change │
│ QQuery 5     │                          766.43ms │                            661.07ms │  +1.16x faster │
│ QQuery 6     │                          474.20ms │                            317.05ms │  +1.50x faster │
│ QQuery 7     │                         1190.87ms │                           1556.33ms │   1.31x slower │
│ QQuery 8     │                          584.23ms │                            743.68ms │   1.27x slower │
│ QQuery 9     │                         1150.48ms │                           1581.38ms │   1.37x slower │
│ QQuery 10    │                          898.01ms │                            875.53ms │      no change │
│ QQuery 11    │                          113.16ms │                            115.31ms │      no change │
│ QQuery 12    │                          321.89ms │                            665.48ms │   2.07x slower │
│ QQuery 13    │                          340.09ms │                            350.92ms │      no change │
│ QQuery 14    │                          566.53ms │                             49.00ms │ +11.56x faster │
│ QQuery 15    │                          123.92ms │                            159.39ms │   1.29x slower │
│ QQuery 16    │                           93.89ms │                            104.72ms │   1.12x slower │
│ QQuery 17    │                          912.10ms │                            842.42ms │  +1.08x faster │
│ QQuery 18    │                         4680.46ms │                           4226.43ms │  +1.11x faster │
│ QQuery 19    │                          857.78ms │                            825.55ms │      no change │
│ QQuery 20    │                          241.99ms │                            353.86ms │   1.46x slower │
│ QQuery 21    │                         1915.38ms │                           1949.02ms │      no change │
│ QQuery 22    │                           90.42ms │                             98.60ms │   1.09x slower │
└──────────────┴───────────────────────────────────┴─────────────────────────────────────┴────────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┓
┃ Benchmark Summary                                  ┃            ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━┩
│ Total Time (on-demand-repartition-with-config)     │ 17155.98ms │
│ Total Time (on-demand-not-always-add-roundrobin)   │ 17357.64ms │
│ Average Time (on-demand-repartition-with-config)   │   779.82ms │
│ Average Time (on-demand-not-always-add-roundrobin) │   788.98ms │
│ Queries Faster                                     │          5 │
│ Queries Slower                                     │         10 │
│ Queries with No Change                             │          7 │
└────────────────────────────────────────────────────┴────────────┘
--------------------
Benchmark tpch_sf1.json
--------------------
┏━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query        ┃ on-demand-repartition-with-config ┃ on-demand-not-always-add-roundrobin ┃        Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 1     │                          106.51ms │                             95.95ms │ +1.11x faster │
│ QQuery 2     │                           20.61ms │                             19.91ms │     no change │
│ QQuery 3     │                           43.59ms │                             36.34ms │ +1.20x faster │
│ QQuery 4     │                           26.63ms │                             22.75ms │ +1.17x faster │
│ QQuery 5     │                           66.74ms │                             54.61ms │ +1.22x faster │
│ QQuery 6     │                           20.47ms │                             17.84ms │ +1.15x faster │
│ QQuery 7     │                           84.02ms │                             74.00ms │ +1.14x faster │
│ QQuery 8     │                           54.67ms │                             49.70ms │ +1.10x faster │
│ QQuery 9     │                           78.47ms │                             66.34ms │ +1.18x faster │
│ QQuery 10    │                           67.30ms │                             59.63ms │ +1.13x faster │
│ QQuery 11    │                           16.11ms │                             14.86ms │ +1.08x faster │
│ QQuery 12    │                           42.73ms │                             33.38ms │ +1.28x faster │
│ QQuery 13    │                           38.63ms │                             32.24ms │ +1.20x faster │
│ QQuery 14    │                           31.97ms │                             29.82ms │ +1.07x faster │
│ QQuery 15    │                           47.89ms │                             42.21ms │ +1.13x faster │
│ QQuery 16    │                           16.23ms │                             14.64ms │ +1.11x faster │
│ QQuery 17    │                          103.78ms │                             95.82ms │ +1.08x faster │
│ QQuery 18    │                          134.12ms │                            117.78ms │ +1.14x faster │
│ QQuery 19    │                           52.97ms │                             48.17ms │ +1.10x faster │
│ QQuery 20    │                           46.33ms │                             39.97ms │ +1.16x faster │
│ QQuery 21    │                          110.85ms │                             93.35ms │ +1.19x faster │
│ QQuery 22    │                           18.11ms │                             16.01ms │ +1.13x faster │
└──────────────┴───────────────────────────────────┴─────────────────────────────────────┴───────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━┓
┃ Benchmark Summary                                  ┃           ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━┩
│ Total Time (on-demand-repartition-with-config)     │ 1228.72ms │
│ Total Time (on-demand-not-always-add-roundrobin)   │ 1075.33ms │
│ Average Time (on-demand-repartition-with-config)   │   55.85ms │
│ Average Time (on-demand-not-always-add-roundrobin) │   48.88ms │
│ Queries Faster                                     │        21 │
│ Queries Slower                                     │         0 │
│ Queries with No Change                             │         1 │
└────────────────────────────────────────────────────┴───────────┘
--------------------
Benchmark tpch_sf10.json
--------------------
┏━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query        ┃ on-demand-repartition-with-config ┃ on-demand-not-always-add-roundrobin ┃        Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 1     │                          843.22ms │                            943.27ms │  1.12x slower │
│ QQuery 2     │                          123.24ms │                            136.28ms │  1.11x slower │
│ QQuery 3     │                          407.30ms │                            446.22ms │  1.10x slower │
│ QQuery 4     │                          198.17ms │                            231.51ms │  1.17x slower │
│ QQuery 5     │                          604.48ms │                            671.58ms │  1.11x slower │
│ QQuery 6     │                          136.38ms │                            158.40ms │  1.16x slower │
│ QQuery 7     │                          887.74ms │                            967.93ms │  1.09x slower │
│ QQuery 8     │                          628.50ms │                            698.72ms │  1.11x slower │
│ QQuery 9     │                         1009.86ms │                           1097.19ms │  1.09x slower │
│ QQuery 10    │                          570.20ms │                            559.98ms │     no change │
│ QQuery 11    │                           90.48ms │                             88.46ms │     no change │
│ QQuery 12    │                          299.50ms │                            282.05ms │ +1.06x faster │
│ QQuery 13    │                          421.75ms │                            421.93ms │     no change │
│ QQuery 14    │                          231.66ms │                            231.48ms │     no change │
│ QQuery 15    │                          384.69ms │                            412.02ms │  1.07x slower │
│ QQuery 16    │                           96.85ms │                             96.86ms │     no change │
│ QQuery 17    │                         1088.84ms │                           1088.24ms │     no change │
│ QQuery 18    │                         1874.24ms │                           1587.75ms │ +1.18x faster │
│ QQuery 19    │                          462.80ms │                            395.33ms │ +1.17x faster │
│ QQuery 20    │                          429.06ms │                            378.36ms │ +1.13x faster │
│ QQuery 21    │                         1564.34ms │                           1344.72ms │ +1.16x faster │
│ QQuery 22    │                          144.52ms │                            130.80ms │ +1.10x faster │
└──────────────┴───────────────────────────────────┴─────────────────────────────────────┴───────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┓
┃ Benchmark Summary                                  ┃            ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━┩
│ Total Time (on-demand-repartition-with-config)     │ 12497.84ms │
│ Total Time (on-demand-not-always-add-roundrobin)   │ 12369.07ms │
│ Average Time (on-demand-repartition-with-config)   │   568.08ms │
│ Average Time (on-demand-not-always-add-roundrobin) │   562.23ms │
│ Queries Faster                                     │          6 │
│ Queries Slower                                     │         10 │
│ Queries with No Change                             │          6 │
└────────────────────────────────────────────────────┴────────────┘

ok, let's revert it then, and we'll see later how it could be improved in other ways